package com.atguigu.flink.chapter11.time;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneOffset;

/**
 * @Author lzc
 * @Date 2022/11/1 08:48
 */
public class Flink_02_Time_Event_1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
       
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
        tEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(0));
        
        // 2. 在 ddl sql 中
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int, " +
                            // 定义事件时间, 一定要定义水印
                            " et as to_timestamp_ltz(ts, 3), " +
                            " watermark for et as et - interval '3' second " + // 需要自己把一个long 转成一个TIMESTAMP(3)   bigint  -> TIMESTAMP(3)
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/sensor.txt', " +
                            " 'format' = 'csv' " +
                            ")");
    
        Table table = tEnv.from("sensor");
        table.printSchema();
        
        tEnv.sqlQuery("select * from sensor").execute().print();
    }
}
